package com.xb.cloud.common.mongo.client;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;

import org.apache.commons.lang3.StringUtils;
import org.mongodb.morphia.DatastoreImpl;
import org.mongodb.morphia.Key;
import org.mongodb.morphia.Morphia;
import org.mongodb.morphia.mapping.Mapper;
import org.mongodb.morphia.query.Query;
import org.mongodb.morphia.query.QueryImpl;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

import com.mongodb.DBObject;
import com.mongodb.Mongo;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.MongoCredential;
import com.mongodb.MongoURI;
import com.mongodb.ReadPreference;
import com.mongodb.ServerAddress;
import com.mongodb.WriteConcern;
import com.mongodb.WriteResult;
import com.mongodb.util.JSON;

/**
 * Mongo 操作实现类
 *
 */
public class SimpleMongoClientImpl implements SimpleMongoClient, InitializingBean, DisposableBean {
	private DatastoreImpl ds;
	private String dbName;
	private String mongoUrl;
	private String userName;
	private String passWord;
	private Boolean needTrace = Boolean.valueOf(false);
	// protected IPersistenceCallback persistenceCallback = new
	// DefaultPersistenceCallback();
	// FailureMonitorJob failureMonitorJob = null;
	private ConcurrentMap<ScheduledFuture<?>, ScheduledExecutorService> schedulerFutures = new ConcurrentHashMap();
	private final transient org.slf4j.Logger logger = LoggerFactory.getLogger(SimpleMongoClientImpl.class);
	private boolean isInitiated = false;
	protected String detectingClassName;
	protected Object detectingId;
	protected long monitorPeriod = 60000L;
	protected int initialDelay = 0;
	protected long detectingTimeoutThreshold = 15000L;
	protected long recheckInterval = 3000L;
	protected int recheckTimes = 3;
	protected boolean positiveTestEnable = false;
	protected boolean isAvailable = true;
	protected WriteConcern defaultWriteConcern = WriteConcern.ACKNOWLEDGED;
	protected boolean slaveOk = true;
	@Deprecated
	protected Boolean safe = null;
	@Deprecated
	protected Boolean fsync = null;
	protected Integer socketTimeout = Integer.valueOf(5000);
	protected Integer connectionsPerHost = null;
	@Deprecated
	protected Boolean autoConnectRetry = null;
	protected Integer connectTimeout = Integer.valueOf(5000);
	protected Integer maxWaitTime = Integer.valueOf(5000);
	protected Boolean keepAlive = null;
	@Deprecated
	protected Long maxAutoConnectRetryTime = null;
	protected Integer threadsAllowedToBlockForConnectionMultiplier = null;
	protected int updaterIntervalMS = 10000;
	protected int updaterIntervalNoMasterMS = 10000;
	protected Integer serverSelectionTimeout = Integer.valueOf(3000);
	private long longTimeRunningSqlIntervalThreshold;

	public void setUpdaterIntervalMS(int updaterIntervalMS) {
		this.updaterIntervalMS = updaterIntervalMS;
	}

	public void setUpdaterIntervalNoMasterMS(int updaterIntervalNoMasterMS) {
		this.updaterIntervalNoMasterMS = updaterIntervalNoMasterMS;
	}

	public void setSocketTimeout(Integer socketTimeout) {
		this.socketTimeout = socketTimeout;
	}

	public void setConnectionsPerHost(Integer connectionsPerHost) {
		this.connectionsPerHost = connectionsPerHost;
	}

	public void setAutoConnectRetry(Boolean autoConnectRetry) {
		this.autoConnectRetry = autoConnectRetry;
	}

	public void setConnectTimeout(Integer connectTimeout) {
		this.connectTimeout = connectTimeout;
	}

	public void setMaxWaitTime(Integer maxWaitTime) {
		this.maxWaitTime = maxWaitTime;
	}

	public void setKeepAlive(Boolean keepAlive) {
		this.keepAlive = keepAlive;
	}

	public void setSlaveOk(Boolean slaveOk) {
		this.slaveOk = slaveOk.booleanValue();
	}

	public void setSafe(Boolean safe) {
		this.safe = safe;
	}

	public void setFsync(Boolean fsync) {
		this.fsync = fsync;
	}

	public void setMaxAutoConnectRetryTime(Long maxAutoConnectRetryTime) {
		this.maxAutoConnectRetryTime = maxAutoConnectRetryTime;
	}

	public void setThreadsAllowedToBlockForConnectionMultiplier(Integer threadsAllowedToBlockForConnectionMultiplier) {
		this.threadsAllowedToBlockForConnectionMultiplier = threadsAllowedToBlockForConnectionMultiplier;
	}

	/*
	 * public void setPersistenceCallback(IPersistenceCallback
	 * persistenceCallback) { this.persistenceCallback = persistenceCallback; }
	 */
	public void setDs(DatastoreImpl ds) {
		this.ds = ds;
	}

	public void setDbName(String dbName) {
		this.dbName = dbName;
	}

	public void setMongoUrl(String mongoUrl) {
		this.mongoUrl = mongoUrl;
	}

	public void setUsername(String userName) {
		this.userName = userName;
	}

	public void setPassword(String passWord) {
		this.passWord = passWord;
	}

	public Boolean getNeedTrace() {
		return this.needTrace;
	}

	public void setNeedTrace(Boolean needTrace) {
		this.needTrace = needTrace;
	}

	public void setServerSelectionTimeout(Integer serverSelectionTimeout) {
		this.serverSelectionTimeout = serverSelectionTimeout;
	}

	public SimpleMongoClientImpl() {
	}

	public SimpleMongoClientImpl(MongoClient mongoClient, String dbName) {
		this.dbName = dbName;
		initDS(new Morphia(), mongoClient, dbName);
	}

	public long getLongTimeRunningSqlIntervalThreshold() {
		return this.longTimeRunningSqlIntervalThreshold;
	}

	public void setLongTimeRunningSqlIntervalThreshold(long longTimeRunningSqlIntervalThreshold) {
		this.longTimeRunningSqlIntervalThreshold = longTimeRunningSqlIntervalThreshold;
	}

	public SimpleMongoClientImpl(MongoClient mongoClient, Mapper map, String dbName) {
		this.dbName = dbName;
		initDS(new Morphia(), map, mongoClient, dbName);
	}

	protected void initDS(Morphia mor, MongoClient mon, String db) {
		this.ds = new DatastoreImpl(mor, mon, db);
	}

	protected void initDS(Morphia mor, Mapper mapper, MongoClient mongoClient, String db) {
		this.ds = new DatastoreImpl(mor, mapper, mongoClient, db);
	}

	private void checkMongoStatus() {
		initMongo();
		if (!this.isInitiated) {
			throw new IllegalStateException("Failed to initiate mongo client");
		}
	}

	public <T> Key<T> save(T entity) {
		return save(entity, this.defaultWriteConcern);
	}

	/*
	 * public <T> Key<T> save(T entity, WriteConcern wc) { if
	 * (!this.isInitiated) { checkMongoStatus(); }
	 * this.persistenceCallback.preSave(entity); Key<T> key =
	 * this.ds.save(entity, wc); this.persistenceCallback.postSave(entity);
	 * 
	 * return key; }
	 */

	public <T> Iterable<Key<T>> save(Iterable<T> entities) {
		return this.ds.save(entities, this.defaultWriteConcern);
	}

	/*
	 * public <T> Iterable<Key<T>> save(Iterable<T> entities, WriteConcern wc) {
	 * if (!this.isInitiated) { checkMongoStatus(); }
	 * this.persistenceCallback.preSave(entities); Iterable<Key<T>> keys =
	 * this.ds.save(entities, wc); this.persistenceCallback.postSave(entities);
	 * 
	 * return keys; }
	 */

	public <T> Key<T> merge(T entity) {
		return merge(entity, this.defaultWriteConcern);
	}

	/*
	 * public <T> Key<T> merge(T entity, WriteConcern wc) { if
	 * (!this.isInitiated) { checkMongoStatus(); }
	 * this.persistenceCallback.preMerge(entity); Key<T> key =
	 * this.ds.merge(entity, wc); this.persistenceCallback.postMerge(entity);
	 * 
	 * return key; }
	 */

	public <T> WriteResult delete(T entity) {
		return delete(entity, this.defaultWriteConcern);
	}

	/*
	 * public <T> WriteResult delete(T entity, WriteConcern wc) { if
	 * (!this.isInitiated) { checkMongoStatus(); }
	 * this.persistenceCallback.preDelete(entity); WriteResult result =
	 * this.ds.delete(entity, wc); this.persistenceCallback.postDelete(entity);
	 * 
	 * return result; }
	 */

	public <T, V> WriteResult deleteById(Class<T> entityClass, V id) {
		return deleteById(entityClass, id, this.defaultWriteConcern);
	}

	/*
	 * public <T, V> WriteResult deleteById(Class<T> entityClass, V id,
	 * WriteConcern wc) { if (!this.isInitiated) { checkMongoStatus(); }
	 * this.persistenceCallback.preDelete(entityClass, id); WriteResult result =
	 * this.ds.delete(entityClass, id, wc);
	 * this.persistenceCallback.postDelete(entityClass, id);
	 * 
	 * return result; }
	 */

	public <T> void ensureIndexes(Class<T> clazz) {
		this.ds.ensureIndexes(clazz);
	}

	public <T> WriteResult deleteByCriteria(QueryImpl<T> query) {
		return deleteByCriteria(query, this.defaultWriteConcern);
	}

	/*
	 * public <T> WriteResult deleteByCriteria(Query<T> query, WriteConcern wc)
	 * { if (!this.isInitiated) { checkMongoStatus(); }
	 * this.persistenceCallback.preDelete(query); WriteResult result =
	 * this.ds.delete(query, wc); this.persistenceCallback.postDelete(query);
	 * 
	 * return result; }
	 */

	public <T, V> T findById(Class<T> entityClass, V id) {
		if (!this.isInitiated) {
			checkMongoStatus();
		}
		return (T) this.ds.get(entityClass, id);
	}

	public <T, V> T findByIdForPrimary(Class<T> entityClass, V id) {
		if (!this.isInitiated) {
			checkMongoStatus();
		}
		this.ds.getMongo().setReadPreference(ReadPreference.primaryPreferred());
		return (T) this.ds.get(entityClass, id);
	}

	public <T> List<T> find(Class<T> entityClass, Map<String, Object> filters) {
		return find(entityClass, filters, null, null, null, null);
	}

	public <T> List<T> find(Class<T> entityClass, Map<String, Object> filters, boolean slaveOk) {
		return find(entityClass, filters, null, null, null, null, slaveOk);
	}

	public <T> List<T> find(Class<T> entityClass, Map<String, Object> filters, Map<String, Boolean> retrievedFields) {
		return find(entityClass, filters, retrievedFields, null, null, null);
	}

	public <T> List<T> find(Class<T> entityClass, Map<String, Object> filters, Map<String, Boolean> retrievedFields,
			boolean slaveOk) {
		return find(entityClass, filters, retrievedFields, null, null, null, slaveOk);
	}

	public <T> List<T> find(Class<T> entityClass, Map<String, Object> filters, Map<String, Boolean> retrievedFields,
			String sort) {
		return find(entityClass, filters, retrievedFields, sort, null, null);
	}

	public <T> List<T> find(Class<T> entityClass, Map<String, Object> filters, Map<String, Boolean> retrievedFields,
			String sort, boolean slaveOk) {
		return find(entityClass, filters, retrievedFields, sort, null, null, slaveOk);
	}

	public <T> List<T> findByExample(T ex) {
		if (!this.isInitiated) {
			checkMongoStatus();
		}
		return this.ds.queryByExample(ex).asList();
	}

	public <T> List<T> find(Class<T> entityClass, Map<String, Object> filters, Map<String, Boolean> retrievedFields,
			String sort, Integer offset, Integer limit) {
		return find(entityClass, filters, retrievedFields, sort, offset, limit, this.slaveOk);
	}

	public <T> List<T> find(Class<T> entityClass, Map<String, Object> filters, Map<String, Boolean> retrievedFields,
			String sort, Integer offset, Integer limit, boolean slaveOk) {
		if (!this.isInitiated) {
			checkMongoStatus();
		}
		Query q = this.ds.createQuery(entityClass);
		if (filters != null) {
			for (String k : filters.keySet()) {
				q.filter(k, filters.get(k));
			}
		}
		if (retrievedFields != null) {
			List<String> included = new ArrayList();
			List<String> excluded = new ArrayList();
			for (String fk : retrievedFields.keySet()) {
				Boolean value = (Boolean) retrievedFields.get(fk);
				if (value != null) {
					if (value.booleanValue()) {
						included.add(fk);
					} else {
						excluded.add(fk);
					}
				}
			}
			if (included.size() > 0) {
				q.retrievedFields(true, (String[]) included.toArray(new String[0]));
			}
			if (excluded.size() > 0) {
				q.retrievedFields(false, (String[]) excluded.toArray(new String[0]));
			}
		}
		if ((sort != null) && (!"".equals(sort))) {
			q.order(sort);
		}
		if (limit != null) {
			q.limit(limit.intValue());
		}
		auditSqlIfNecessary("query", entityClass.getName(), null, null, this.dbName, 0L, getCurrentTime());
		return q.asList();
	}

	public <T> Query<T> getQuery(Class<T> entityClass) {
		if (!this.isInitiated) {
			checkMongoStatus();
		}
		return this.ds.createQuery(entityClass);
	}

	public <T> List<T> queryByCriteria(Query<T> query) {
		if (!this.isInitiated) {
			checkMongoStatus();
		}
		return query.asList();
	}

	private String getCurrentTime() {
		SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm");
		return df.format(new Date());
	}

	public <T> T jsonToEntity(Class<T> entityClass, String json) {
		if (!this.isInitiated) {
			checkMongoStatus();
		}
		return (T) this.ds.getMapper().fromDBObject(this.ds, entityClass, (DBObject) JSON.parse(json),
				this.ds.getMapper().createEntityCache());
	}

	public <T> String entityToJson(T entity) {
		if (!this.isInitiated) {
			checkMongoStatus();
		}
		return this.ds.getMapper().toDBObject(entity).toString();
	}

	protected void auditSqlIfNecessary(String opType, String statementName, Object parameterObject, String errorMsg,
			String dsName, long interval, String exeTime) {
	}

	private void initMongo() {
		if ((this.mongoUrl == null) || (this.dbName == null) || (this.userName == null) || (this.passWord == null)) {
			throw new IllegalArgumentException(
					"The 'mongoUrl'or 'dbName' or 'userName' or 'passWord' property must be set");
		}
		/*
		if (this.persistenceCallback == null) {
			throw new IllegalArgumentException("The 'persistenceCallback' property must be set");
		}
		*/
		try {
			System.setProperty("com.mongodb.updaterIntervalMS", this.updaterIntervalMS + "");
			System.setProperty("com.mongodb.updaterIntervalNoMasterMS", this.updaterIntervalNoMasterMS + "");
		} catch (SecurityException e) {
			this.logger.error("", e);
		}
		MongoClientOptions.Builder builder = new MongoClientOptions.Builder();
		if (this.socketTimeout != null) {
			builder.socketTimeout(this.socketTimeout.intValue());
		}
		if (this.connectionsPerHost != null) {
			builder.connectionsPerHost(this.connectionsPerHost.intValue());
		}
		if (this.connectTimeout != null) {
			builder.connectTimeout(this.connectTimeout.intValue());
		}
		if (this.maxWaitTime != null) {
			builder.maxWaitTime(this.maxWaitTime.intValue());
		}
		if (this.keepAlive != null) {
			builder.socketKeepAlive(this.keepAlive.booleanValue());
		}
		if (this.threadsAllowedToBlockForConnectionMultiplier != null) {
			builder.threadsAllowedToBlockForConnectionMultiplier(
					this.threadsAllowedToBlockForConnectionMultiplier.intValue());
		}
		if (this.serverSelectionTimeout != null) {
			builder.serverSelectionTimeout(this.serverSelectionTimeout.intValue());
		}
		try {
			MongoClientOptions option = builder.build();
			MongoURI uri = new MongoURI(this.mongoUrl);
			List<ServerAddress> replicaSetSeeds = new ArrayList(uri.getHosts().size());
			for (String host : uri.getHosts()) {
				replicaSetSeeds.add(new ServerAddress(host));
			}
			MongoCredential mongoCredent = MongoCredential.createCredential(this.userName, this.dbName,
					this.passWord.toCharArray());
			List<MongoCredential> mongoCredentList = new ArrayList();
			mongoCredentList.add(mongoCredent);
			MongoClient mongoClient = new MongoClient(replicaSetSeeds, mongoCredentList, option);
			if (this.slaveOk) {
				mongoClient.setReadPreference(ReadPreference.primaryPreferred());
			} else {
				mongoClient.setReadPreference(ReadPreference.secondaryPreferred());
			}
			if ((StringUtils.isNotEmpty(this.dbName)) || (mongoClient != null)) {
				initDS(new Morphia(), mongoClient, this.dbName);
			}
			//MongoStatus.setAvailable(true);
			this.isInitiated = true;
		} catch (Exception e) {
			//MongoStatus.setAvailable(false);
			this.isInitiated = false;
			this.logger.error("", e);
		}
	}

	public void afterPropertiesSet() throws Exception {
		java.util.logging.Logger.getLogger("com.mongodb.ReplicaSetStatus").setLevel(Level.SEVERE);

		initMongo();
		if (this.positiveTestEnable) {
			//this.failureMonitorJob = new FailureMonitorJob(this);
			//this.failureMonitorJob.setDetectingRequestTimeout(this.detectingTimeoutThreshold);
			try {
				Class detectingClass = Class.forName(this.detectingClassName);
				//this.failureMonitorJob.setDetectingClass(detectingClass);
			} catch (ClassNotFoundException e) {
				throw new IllegalArgumentException(
						"The 'detectingClassName' class not found:" + this.detectingClassName);
			}
			//this.failureMonitorJob.setDetectingId(this.detectingId);
			//this.failureMonitorJob.setRecheckInterval(this.recheckInterval);
			//this.failureMonitorJob.setRecheckTimes(this.recheckTimes);

			ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
			/*
			ScheduledFuture<?> future = scheduler.scheduleWithFixedDelay(this.failureMonitorJob, getInitialDelay(),
					getMonitorPeriod(), TimeUnit.MILLISECONDS);
			*/
			//this.schedulerFutures.put(future, scheduler);
		}
	}

	public void destroy() throws Exception {
		if (this.ds != null) {
			Mongo mongo = this.ds.getMongo();
			if (mongo != null) {
				mongo.close();
			}
		}
		/*
		if (this.positiveTestEnable) {
			for (Map.Entry<ScheduledFuture<?>, ScheduledExecutorService> e : this.schedulerFutures.entrySet()) {
				ScheduledFuture<?> future = (ScheduledFuture) e.getKey();
				ScheduledExecutorService scheduler = (ScheduledExecutorService) e.getValue();
				future.cancel(true);
				shutdownExecutor(scheduler);
			}
			shutdownExecutor(this.failureMonitorJob.getExecutor());
		}
		*/
	}

	private void shutdownExecutor(ExecutorService executor) {
		try {
			executor.shutdown();
			executor.awaitTermination(5L, TimeUnit.SECONDS);
		} catch (Exception ex) {
			this.logger.warn("interrupted when shutting down executor service.");
		}
	}

	public void setAvailable(boolean isAvailable) {
		this.isAvailable = isAvailable;
		//MongoStatus.setAvailable(isAvailable);
	}

	public boolean isPositiveTestEnable() {
		return this.positiveTestEnable;
	}

	public void setPositiveTestEnable(boolean positiveTestEnable) {
		this.positiveTestEnable = positiveTestEnable;
	}

	public long getMonitorPeriod() {
		return this.monitorPeriod;
	}

	public void setMonitorPeriod(long monitorPeriod) {
		this.monitorPeriod = monitorPeriod;
	}

	public int getInitialDelay() {
		return this.initialDelay;
	}

	public void setInitialDelay(int initialDelay) {
		this.initialDelay = initialDelay;
	}

	public long getDetectingTimeoutThreshold() {
		return this.detectingTimeoutThreshold;
	}

	public void setDetectingTimeoutThreshold(long detectingTimeoutThreshold) {
		this.detectingTimeoutThreshold = detectingTimeoutThreshold;
	}

	public long getRecheckInterval() {
		return this.recheckInterval;
	}

	public void setRecheckInterval(long recheckInterval) {
		this.recheckInterval = recheckInterval;
	}

	public int getRecheckTimes() {
		return this.recheckTimes;
	}

	public void setRecheckTimes(int recheckTimes) {
		this.recheckTimes = recheckTimes;
	}

	public String getDetectingClassName() {
		return this.detectingClassName;
	}

	public void setDetectingClassName(String detectingClassName) {
		this.detectingClassName = detectingClassName;
	}

	public Object getDetectingId() {
		return this.detectingId;
	}

	public void setDetectingId(Object detectingId) {
		this.detectingId = detectingId;
	}

	@Override
	public <T> Key<T> save(T entity, WriteConcern wc) {
		// TODO Auto-generated method stub
		return this.ds.save(entity, this.defaultWriteConcern);
	}

	@Override
	public <T> Key<T> merge(T entity, WriteConcern wc) {
		// TODO Auto-generated method stub
		return this.ds.merge(entity, this.defaultWriteConcern);
	}

	@Override
	public <T> WriteResult delete(T entity, WriteConcern wc) {
		// TODO Auto-generated method stub
		return this.ds.delete(entity, this.defaultWriteConcern);
	}

	@Override
	public <T, V> WriteResult deleteById(Class<T> entityClass, V id, WriteConcern wc) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public <T> WriteResult deleteByCriteria(Query<T> query, WriteConcern wc) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public <T> WriteResult deleteByCriteria(Query<T> query) {
		// TODO Auto-generated method stub
		return null;
	}
}
